#include "log/logfile.hxx"
#include "util/serialize.hxx"
#include "util/io.hxx"
#include "util/fs.hxx"
#include "util/common.hxx"
#include "util/crc32.hxx"
#include "util/logger.hxx"
#include "db/metafile.hxx"
#include <assert.h>


uint64_t LogRecord::MAGIC = 0x31422413567890;
Dbt* LogRecord::m_serialBuffer = NULL;

uint32_t LogRecord::calculateRecordLength() {

    uint32_t len =  sizeof(m_magic) + 
        sizeof(m_scn) + sizeof(m_type) + 
        sizeof(m_index) + sizeof(m_addrvalueNew) + sizeof(m_addrvalueOld) +
        m_dataStoreName.size() + sizeof(uint32_t) +
        m_dbPartitionName.size() + sizeof(uint32_t) +
        sizeof(m_loadSizeOld) + sizeof(m_loadSizeNew) +
        m_dataNew.getDatalen() + sizeof(uint32_t) + sizeof(m_crc);

    return len;
}    

std::string LogRecord::toString() {

    InfoSerializer ser;

    ser.appendInt("length", m_length);
    ser.appendLong("magic", m_magic);
    ser.appendLong("scn", m_scn);
    ser.appendInt("type",m_type);

    ser.appendInt("loadSizeOld", m_loadSizeOld);
    ser.appendInt("loadSizeNew", m_loadSizeNew);

    ser.appendInt("index", m_index);
    ser.appendLong("addrvalueNew", m_addrvalueNew);
    ser.appendLong("addrvalueOld", m_addrvalueOld);

    ser.appendString("store", m_dataStoreName);
    ser.appendString("partition", m_dbPartitionName);

    ser.appendDbt("data", m_dataNew);

    ser.appendInt("crc", m_crc);

    return ser.str();    
}

void LogRecord::adjustSerialBuffer(uint32_t len) {
    
    uint32_t initialSize = 32*1024;
    if(LogRecord::m_serialBuffer) initialSize = LogRecord::m_serialBuffer->getSize();

    for(; initialSize < len; ) {
    
        initialSize = (initialSize << 1);
    }

    if(LogRecord::m_serialBuffer == NULL) {

        char* buf = new char[initialSize];
        LogRecord::m_serialBuffer = new Dbt(buf, initialSize, 0, DBT_MALLOC);
    }
    else if(initialSize != LogRecord::m_serialBuffer->getSize()) {

        char* buf = new char[initialSize];
        delete[] LogRecord::m_serialBuffer->data();
        LogRecord::m_serialBuffer->setBuffer(buf, initialSize, 0, DBT_MALLOC);
    }
}

Dbt* LogRecord::serialize() {

    m_length = calculateRecordLength();

    adjustSerialBuffer(m_length);

    LogRecord::m_serialBuffer->setDatalen(0);
    DbtSerializer ser(LogRecord::m_serialBuffer);


    ser.appendInt(m_length);
    ser.appendLong(m_magic);
    ser.appendLong(m_scn);
    ser.appendInt(m_type);

    ser.appendInt(m_loadSizeOld);
    ser.appendInt(m_loadSizeNew);

    ser.appendInt(m_index);
    ser.appendLong(m_addrvalueNew);
    ser.appendLong(m_addrvalueOld);

    ser.appendString(m_dataStoreName);
    ser.appendString(m_dbPartitionName);

    ser.appendDbt(m_dataNew);


    //CPMACT record doedn't include critical data, so skip crc checksum
    m_crc = 0;
    if(m_type != LogRecord::COMPACT) {
        //calculate crc32
        CRC32 crc; 
        crc.update(ser.getDbt()->data(), 0, ser.getDbt()->getDatalen());
        m_crc = crc.getValue();
    }
    ser.appendInt(m_crc);

	return ser.getDbt();
}


//data doesn't include length, it starts with maigc
bool LogRecord::deserialize(const std::string& data) {

    Deserializer des(data);
	//m_length = des.readInt();
    m_magic = des.readLong();
	m_scn = des.readLong();
	m_type = des.readInt();
    m_loadSizeOld = des.readInt();
    m_loadSizeNew = des.readInt();
    m_index = des.readInt();
    m_addrvalueNew = des.readLong();
    m_addrvalueOld = des.readLong();

	uint32_t len = 0;
	
	len = des.readInt();
	m_dataStoreName = des.readString(len);

	len = des.readInt();
	m_dbPartitionName = des.readString(len);

	len = des.readInt();
    if(len > 0) {
        char* buf = new char[len];
        m_dataNew.setBuffer(buf, len, 0, DBT_MALLOC); 
        des.readDbt(len, m_dataNew);
    }

	m_crc = des.readInt();

	return true;
	
}

uint64_t LogFileHeader::MAGIC=0x1234567890;

LogFileHeader::LogFileHeader()
: m_version(0), 
  m_magic(LogFileHeader::MAGIC) {

}    

std::string LogFileHeader::serialize() {

    StringSerializer ser;

	ser.appendLong(m_version);
	ser.appendLong(m_magic);

    return ser.str();
}


void LogFileHeader::deserialize(const std::string& data) {

	Deserializer des((char*)data.c_str(), data.size());

	m_version = des.readLong();
	m_magic = des.readLong();
}

std::string LogFileHeader::toString() {

    InfoSerializer ser;

    ser.appendLong("version", m_version);
    ser.appendLong("magic", m_magic);

    return ser.str();
}

bool LogFileHeader::checkHeader() {

    return m_magic == LogFileHeader::MAGIC; 
}


uint32_t LogFile::HEADER_SIZE = 1024;

LogFile::LogFile(const std::string logName, uint32_t  logId)
: m_fileName(logName), 
  m_logId(logId), 
  m_raf(NULL) {
	
}

LogFile::~LogFile() {

    if(m_raf) { delete m_raf; m_raf = NULL; }
}

DbStatus LogFile::init() {

    DbStatus ret;

    m_raf = new PosixRandomAccessFile(m_fileName);
    ret = m_raf->init(); if(!ret.isok()) return ret;
}


DbStatus LogFile::saveHeader() {

    DbStatus ret;

    std::string header = m_header.serialize();
    ret = m_raf->write(0, header);
    if(!ret.isok()) return ret;

//    ret = m_raf->flush();
    return ret;
}


bool LogFile::isSynced() {

   return (MetaFile::getInstance()->isLogSynced(m_logId) == 1)?true:false; 
}


void LogFile::setSynced() {

    return MetaFile::getInstance()->setLogSync(m_logId, 1); 
}


LogFileWriter::LogFileWriter(uint32_t logId, const std::string& logPath, uint32_t capacity)
: LogFile(logPath+"/"+ utility::itoa(logId) + ".log", logId),
  m_capacity(capacity), 
  m_writer(NULL) {

}


LogFileWriter::~LogFileWriter() {

    if(m_raf) { delete m_raf; m_raf = NULL; }
	if(m_writer) {delete m_writer; m_writer = NULL;}
}



DbStatus LogFileWriter::init() {

	DbStatus ret;

	bool newfile = false;
	if(!FsUtility::fileExists(m_fileName)) {
	
		newfile = true;
	}


	m_writer = new PosixMmapSequentialWriter(m_fileName, getpagesize(), LogFile::HEADER_SIZE, false);
	ret = m_writer->init(); if(!ret.isok()) return ret;

    ret = LogFile::init(); if(!ret.isok()) return ret;

	if(newfile) {
	
		ret = saveHeader();
		
    }
	else {
	
		ret = loadHeader();
        if(ret.getCode() == DbStatus::DB_PARTIALREAD || ret.getCode() ==
                DbStatus::DB_CORRUPTION) {
        
            ret = saveHeader();
        }
	}

	return ret;
}


DbStatus LogFileWriter::close() {

    DbStatus ret;

    if(m_writer) {
    
        ret = m_writer->close();
        if(!ret.isok()) return ret;
    }

    if(m_raf) {
        
        ret = m_raf->close();
        if(!ret.isok()) return ret;
    }

    return ret;
}


DbStatus LogFile::loadHeader() {

	DbStatus ret;
	
	std::string header;
	ret = m_raf->read(0, 1024, header);
	if(!ret.isok()) return ret;

	
	m_header.deserialize(header);
	bool isvalid = m_header.checkHeader();
	if(!isvalid) ret = DbStatus("LogFileWriter::loadHeader", DbStatus::DB_CORRUPTION);

    return ret;	
}	


DbStatus LogFileWriter::flush() {

	return m_writer->flush();
}


DbStatus LogFileWriter::append(std::string& logRec) {

    uint32_t offset = m_writer->getNextOffset();
    if(offset + logRec.size() > m_capacity) 
        return DbStatus("LogFileWriter::append", DbStatus::DB_LOGOVERFLOW);    
	
    return m_writer->append(logRec);
}


DbStatus LogFileWriter::append(Dbt* dbt) {

    uint32_t offset = m_writer->getNextOffset();
    if(offset + dbt->getDatalen() > m_capacity)
        return DbStatus("LogFileWriter::append", DbStatus::DB_LOGOVERFLOW);
    
    return m_writer->append(dbt->data(), dbt->getDatalen());
}

uint32_t LogFileWriter::getNextOffset() {

    return m_writer->getNextOffset(); 
}


DbStatus LogFileWriter::closeWriter() {

    return m_writer->close();
}


LogFileReader::LogFileReader(uint32_t logId, const std::string& logPath)
: LogFile(logPath+"/"+ utility::itoa(logId) + ".log", logId), 
  m_reader(NULL) {

}


LogFileReader::~LogFileReader() {

    if(m_reader) {delete m_reader; m_reader = NULL;}
}

DbStatus LogFileReader::init() {

   DbStatus ret;

   ret = LogFile::init(); if(!ret.isok()) return ret;

   ret = loadHeader(); if(!ret.isok()) return ret;

   m_reader = new PosixSequentialReader(m_fileName);
   ret = m_reader->init();
   if(!ret.isok()) return ret;

   ret = m_reader->skip(LogFile::HEADER_SIZE);
   return ret; 
}

//io error: return error
//io success: 
//(1) partial read:return partial error
//(2) no more data in file: return ok and size=0
//(3) complete read: return ok and actual size
DbStatus LogFileReader::readInt(uint32_t& data) {

    DbStatus ret;

    data = 0;

    std::string result;
    ret = m_reader->read(sizeof(uint32_t), result); if(!ret.isok()) return ret;
    if(result.size() && result.size() != sizeof(uint32_t)) {

        m_reader->skip(0-result.size());
        ret = truncateToCurrentOffset(); if(!ret.isok()) return ret;
        return DbStatus("LogFileReader::readInt", DbStatus::DB_PARTIALREAD);
    }

    if(result.size()) data = *(uint32_t*)result.c_str();

    return ret;
}


DbStatus LogFileReader::getNextLogRecord(bool& found, LogRecord& rec) {
    
    DbStatus ret; 
    found = false;
   
    //philosophy to handle corrupt log record: mydb will be used in a distributed
    //system with replication. If any log corruption happens, mydb will truncate
    //to the last good record and report the truncated data. The replication will 
    //recover the lost data if any. 
    // handle 3 cases:
    //1. length is 0 becaue we reach end of file;
    //2. checksum is not correct;
    //3. partial log record which is caused by last incomplete log writting due
    //to crash or power failure;


    uint32_t length;
    ret = readInt(length); 
    if(!ret.isok()) {
        if(DbStatus::DB_PARTIALREAD == ret.getCode()) { 
    
            ret = DbStatus("LogFileReader::getNextLogRecord", DbStatus::DB_CORRUPTION);
            Logger::log("readInt get an invalid length, logfile: %s",
                    m_fileName.c_str());
        }
        return ret;
    }

    if(length == 0) return ret;
    
    std::string result;
    ret =  m_reader->read(length, result); if(!ret.isok()) return ret;
    if(result.size() != length) {
        
        m_reader->skip(0 - result.size() - sizeof(length));
        ret = truncateToCurrentOffset(); if(!ret.isok()) return ret;
        ret = DbStatus("LogFileReader::getNextLogRecord", DbStatus::DB_CORRUPTION);
        Logger::log("read partial data, logfile: %s", m_fileName.c_str());
        return ret;
    }

    rec.setLength(length);

    rec.deserialize(result);

    if(rec.getType() != LogRecord::COMPACT) {
        //check crc32
        CRC32 crc;
        crc.update(length);
        crc.update(result.c_str(), 0, result.size() - sizeof(uint32_t));
        uint32_t crcValue = crc.getValue(); 
        if(crcValue != rec.getCRC()) {

            m_reader->skip(0 - result.size() - sizeof(length));
            ret = truncateToCurrentOffset(); if(!ret.isok()) return ret;
            Logger::log("log record crc is not matched, logfile: %s, crc: %d, expected crc: %d", 
                    m_fileName.c_str(),
                    crcValue,
                    rec.getCRC());
            return DbStatus("LogFileReader::getNextLogRecord", DbStatus::DB_CORRUPTION);
        }
    }

    found = true;
    
    return ret;
}




//if the last record is partial, return error;else return ok
DbStatus LogFileReader::getAllRecords(std::vector<LogRecord>& records) {

    DbStatus ret;

    while(1) {
    
        LogRecord rec; bool found;
        ret = getNextLogRecord(found, rec);
        if(!ret.isok() || found == false) break;

        records.push_back(rec);
    }

    return ret;
}

DbStatus LogFileReader::truncateToCurrentOffset() {
    DbStatus ret;

    uint64_t offset;
    ret = m_reader->getCurPos(offset); if(!ret.isok()) return ret;
    ret = FsUtility::truncateFile(m_fileName, offset); if(!ret.isok()) return ret;
    return ret;
}

DbStatus LogFileReader::close() {
    
    DbStatus ret;

    if(m_reader) {
    
        ret = m_reader->close();
        if(!ret.isok()) return ret;
    }

    if(m_raf) {
        
        ret = m_raf->close();
        if(!ret.isok()) return ret;
    }

    return ret;
}

